package com.lagou.java;

import com.alibaba.fastjson.JSON;
import com.lagou.java.service.UserService;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;

import java.util.*;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors;

//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

/**
 * @Classname ZkBoostrap
 * @Description TODO
 * @Date 2020/8/20 12:59
 * @Created by kaifa
 */
@Slf4j
public class ZkBoostrap {

//    private static final Logger log = LoggerFactory.getLogger(ZkBoostrap.class);


    // 存放远程提供服务的地址
    public volatile static Set<ServerInfo> remoteServiceAddresses = Collections.synchronizedSet(new HashSet<>());

    // zookeeper 客户端
    private static CuratorFramework zkClient;

    public static Set<String> clientNames = new CopyOnWriteArraySet<>();
//    private final List<UserService> services = new CopyOnWriteArrayList<>();

    // zookeeper 注册中心地址
    private String zkAddress;

    public ZkBoostrap(String zkAddress, List<String> clientNames) {
        ZkBoostrap.clientNames.addAll(clientNames);
        zkAddress = zkAddress;
        this.zkClient = CuratorFrameworkFactory.newClient(zkAddress, new RetryNTimes(2, 2000));
        this.zkClient.start();
        this.initRemoteServicesConnetion(zkAddress);
        this.addListners();
    }

    // 初始化远程连接
    public static void initRemoteServicesConnetion(String zkAddress) {
        try {
            List<String> childrenNodes = zkClient.getChildren().forPath(ZkPathEnum.BASE_SERVER_PATH);
            System.out.println("===>>> zookeeper 初始化客户端连接， 获取到提供服务的远程节点为：" + childrenNodes);
            for (String clientName: clientNames) {
                for (String node : childrenNodes) {
                    String serverInfoStr = new String(zkClient.getData().forPath(ZkPathEnum.BASE_SERVER_PATH + "/" + node));
                    ServerInfo serverInfo = JSON.parseObject(serverInfoStr, ServerInfo.class);
                    // 初始化 netty 客户端
//                    String clientName = RpcConsumer.getClientName();
                    ClientBootstrap.AddressInfo addressInfo = new ClientBootstrap.AddressInfo(clientName, serverInfo.getHost() + ":" + serverInfo.getPort());
                    ClientBootstrap.ClientInfo client = ClientBootstrap.addressesWithHandlers.get(addressInfo);
                    if (client == null) {
                        client = RpcConsumer.initClient(serverInfo.getHost(), serverInfo.getPort());
                        ClientBootstrap.addressesWithHandlers.put(addressInfo, client);
                    }

                    UserService proxy = RpcConsumer.createProxy(
                            UserService.class,
                            serverInfo.getHost(),
                            serverInfo.getPort(),
                            clientName
                    );
                    // 更新 serverInfo
                    remoteServiceAddresses.add(serverInfo);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 监控节点
    private void addListners() {
        try {
            PathChildrenCache childrenListner = new PathChildrenCache(zkClient, ZkPathEnum.BASE_SERVER_PATH, true);
            childrenListner.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    if (event != null && event.getData() != null) {
                        log.info("============================= netty 服务端节点信息发生变更 =============================================");
                        String path = event.getData().getPath();
                        if (path.startsWith(ZkPathEnum.BASE_SERVER_PATH)) {
                            String childNode = path.substring(ZkPathEnum.BASE_CLIENT_PATH.length() + 1);
                            // 新增子节点
                            if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event.getType())) {
                                String nodeVal = new String(event.getData().getData());
                                log.info("新增子节点, 节点为：{}, 详细信息为：{}", childNode, nodeVal);
                            }

                            // 子节点信息发生变更
                            if (PathChildrenCacheEvent.Type.CHILD_UPDATED.equals(event.getType())) {
                                String nodeVal = new String(event.getData().getData());
                                log.info("子节点信息发生变更, 节点为：{}, 详细信息为：{}", childNode, nodeVal);
                            }
                            // 若本地没有服务缓存， 直接退出
                            if(remoteServiceAddresses.size() < 1) return;
                            // 更新子节点信息到本地缓存
                            List<String> childrenNodes = zkClient.getChildren().forPath(ZkPathEnum.BASE_SERVER_PATH);
                            Set<String> addressesStr = remoteServiceAddresses.stream().map(info -> info.getAddress()).collect(Collectors.toSet());
                            for (String node : childrenNodes) {
                                String serverInfoStr = new String(zkClient.getData().forPath(ZkPathEnum.BASE_SERVER_PATH + "/" + node));
                                ServerInfo serverInfo = JSON.parseObject(serverInfoStr, ServerInfo.class);
                                String address = serverInfo == null ? null : serverInfo.getAddress();
                                // 检查是否缓存该地址
                                if (!StringUtil.isNullOrEmpty(address) && !addressesStr.contains(address)) {
                                    // 更新 serverInfo
                                    remoteServiceAddresses.add(serverInfo);
                                    // 创建动态代理服务
                                    for (String clientName : clientNames) {
                                        UserService proxy = RpcConsumer.createProxy(
                                                UserService.class,
                                                serverInfo.getHost(),
                                                serverInfo.getPort(),
                                                clientName
                                        );
                                    }
                                }
                            }

                            // 子节点被删除
                            if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
                                String nodedVal = new String(event.getData().getData());
                                log.warn("子节点被删除, 节点为：{}, 详细信息为：{}", childNode, nodedVal);

                                // 若本地没有服务缓存， 直接退出
                                if(remoteServiceAddresses.size() < 1) return;

                                ServerInfo serverInfo = JSON.parseObject(nodedVal, ServerInfo.class);
                                // 断开所有与该服务器的客户端连接
                                List<ClientBootstrap.AddressInfo> removedKeys = new ArrayList<>();
                                ClientBootstrap.addressesWithHandlers.entrySet().forEach((Map.Entry<ClientBootstrap.AddressInfo, ClientBootstrap.ClientInfo> entry) -> {
                                    if (entry.getKey().hasRemoteAddress(serverInfo.getHost(), serverInfo.getPort())) {
                                        ClientBootstrap.ClientInfo clientInfo = entry.getValue();
                                        clientInfo.channelFuture.channel().close();
                                        clientInfo.group.shutdownGracefully();
                                        removedKeys.add(entry.getKey());
                                    }
                                });
                                removedKeys.forEach(Key -> ClientBootstrap.addressesWithHandlers.remove(Key));
                                remoteServiceAddresses.remove(serverInfo);
                            }
                            log.info("发生改变的的服务节点为：{}", new String(event.getData().getData()));
                        }
                    }

                }
            });
            // 开始监听
            childrenListner.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


}
